/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package com.viettel.vinamilk.memcached;

import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import net.spy.memcached.MemcachedClient;
import net.spy.memcached.internal.GetFuture;
import org.apache.log4j.Logger;

/**
 *
 * @author hienpt9
 */
public class MemcachedClientManager implements IMemcachedClient {

    final String CHECK_AVAILABILITY_KEY = "CHECK_AVAILABILITY_KEY";

    // Logger dung de ghi log. 
    // WARN: Chu y, config cua log4j phai duoc load khi khoi tao
    static Logger logger = Logger.getLogger(MemcachedClientManager.class);

    // Doi tuong quan ly memcached clients
    static MemcachedClientManager instance = null;

    // Mang cac memcached clients
    ArrayList<MemcachedClient> memcachedClients = null;

    // next memcached index for selection arcording to roudrobin
    final Object lock = new Object();
    int nextMemcachedIndex = 0;

    // List memcached server adress
    static String LST_MEMCACHED_SV_ADDRESS = "";

    // state
    static boolean RUNNING = false;

    private MemcachedClientManager() {
        // Logging
        logger.warn("Starting create new MemcachedClientManager.");
        // Khoi tao ket noi toi 2 server memcached
        List<InetSocketAddress> lstServer = new ArrayList<>();

        // Doc dia chi tu file cau hinh, 
        // neu cau hinh sai hoac 1 trong so cac IP khong ket noi duoc throw exception
        //String[] listServer = AppSetting.getStringValue("hostMemCached").split(",");
        String[] listServer = LST_MEMCACHED_SV_ADDRESS.split(",");
        logger.warn("[Step 1 - OK] List memcached server configuration loading is completed.");
        for (String child : listServer) {
            try {
                String[] detail = child.split(":");

                boolean res = isServerValid(detail[0], Integer.parseInt(detail[1]));
                if (res) {
                    InetSocketAddress svAddress = new InetSocketAddress(
                            detail[0], Integer.parseInt(detail[1]));
                    lstServer.add(svAddress);
                }
            } catch (Exception ex) {
                // Log loi
                logger.error("[NOT OK] There are some problems when trying to test InetSocketAddress. MemcachedClientManager create unsuccessfully ", ex);
                return;
            }
        }

        // Tao cac memcached client connect toi tung memcached server
        memcachedClients = new ArrayList<>();
        for (int i = 0; i < lstServer.size(); i++) {
            try {
                MemcachedClient memcachedClient;
                memcachedClient = new MemcachedClient(lstServer.get(i));
                memcachedClients.add(memcachedClient);
            } catch (IOException ex) {
                //Log loi
                logger.error("[NOT OK] There are some problems when trying to create new MemcachedClient[" + lstServer.get(i) + "]. MemcachedClientManager create unsuccessfully ", ex);
                return;
            }
        }
        logger.warn("[Step 2 - OK] All Memcached Clients are created: " + lstServer);
        try {
            // Wait 5s for sure the memcached client has been created successfully.
            while (true) {
                if (isMemcachedUseable()) {
                    break;
                }
                Thread.sleep(1);
            }
        } catch (InterruptedException ex) {
            logger.error("InterruptedException when delays use Thread.sleep().", ex);
        }
        logger.warn("[OK] MemcachedClientManager create successfully.");
    }

    /**
     * Load cau hinh, khoi tao Memcached client manager
     *
     * @param lstAddress
     * @throws Exception
     */
    public static void start(String lstAddress) throws Exception {
        if (!RUNNING) {
            LST_MEMCACHED_SV_ADDRESS = lstAddress;
            getInstance();
            RUNNING = true;
        } else {
            logger.warn("Memcached client manager has been started. Do not start any more!");
        }
    }

    public static synchronized MemcachedClientManager getInstance() throws Exception {
        if (instance == null) {
            instance = new MemcachedClientManager();
        }
        return instance;
    }

    @Override
    public boolean set(String key, int expried, Object value) {
        boolean returnValue = false;
        for (MemcachedClient memcachedClient : memcachedClients) {
            if (memcachedClient.getAvailableServers().size() > 0) {
                try {
                    memcachedClient.set(key, expried, value);
                    returnValue = true;
                    logger.info("Data cached with k = " + key + " on " + memcachedClient.getAvailableServers());
                } catch (Exception e) {
                    logger.warn("Can't set (k=" + key + ",v= *** this value has been disabled for a security reason.) to " + memcachedClient.getAvailableServers(), e);
                }
            } else {
                logger.warn("Can't set (k=" + key + ",v= *** this value has been disabled for a security reason) to " + memcachedClient.getUnavailableServers() + ". Memcached server is not available now.");
            }
        }
        return returnValue;
    }

    @Override
    public Object get(String key, long timeout) {
        Object returnValue = null;
        int numOfTrying = 0;
        boolean hitCached = false;
        //Lay memcached client theo thuat toan roudrobin
        int currentMemcachIndex = getNextMemcachedClientIndex();

        // Khi chua lay duoc key, va so lan trying < tong so memcachedClient
        while (!hitCached && numOfTrying < memcachedClients.size()) {
            // Lay ra memcached client de xu ly
            MemcachedClient memcachClient = getMemcachedClient(currentMemcachIndex, numOfTrying);

            // Neu memcached server khong available, next
            if (memcachClient.getAvailableServers().isEmpty()) {
                numOfTrying++;
                logger.error("[Time trying " + numOfTrying + "] Key=" + key + " Server: " + memcachClient.getUnavailableServers() + " is unavailable!");
                continue;
            }

            // returnValue
            GetFuture<Object> f = memcachClient.asyncGet(key);
            try {
                returnValue = f.get(timeout, TimeUnit.MILLISECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                // Timeout
                f.cancel(true);
                numOfTrying++;
                logger.error("[Time trying " + numOfTrying + "] Can't get key=" + key + " on server " + memcachClient.getAvailableServers()+" | "+memcachClient.getUnavailableServers(), e);
                continue;
            }

            if (returnValue != null) {
                // Hit cached
                hitCached = true;
                logger.info("Cached hit for key = " + key + " on " + memcachClient.getAvailableServers());
            } else {
                numOfTrying++;
                logger.info("[Time trying " + numOfTrying + "] Key=" + key + " is not exist on " + memcachClient.getAvailableServers());
            }
        }
        if (returnValue == null) {
            logger.info("Cached miss for key = " + key);
        }
        return returnValue;
    }

    private boolean isServerValid(String ip, int port) throws Exception {
        try (Socket socket = new Socket()) {
            socket.connect(new InetSocketAddress(ip, port), 1000);
            InputStream is = socket.getInputStream();
            is.close();
        } catch (Exception e) {
            // TODO: Logger.error
            logger.error("Can not connect with Address: " + ip + " : " + port, e);
            throw e;
        }
        return true;
    }

    private synchronized int getNextMemcachedClientIndex() {
        //MemcachedClient memcachedClient = memcachedClients.get(nextMemcachedIndex);
        //System.out.println("Current memcached client: Available="+memcachedClient.getAvailableServers()+"| unavailable="+memcachedClient.getUnavailableServers());
        if (nextMemcachedIndex == (memcachedClients.size() - 1)) {
            nextMemcachedIndex = 0;
        } else {
            nextMemcachedIndex++;
        }
        return nextMemcachedIndex;
    }

    private MemcachedClient getMemcachedClient(int currentMemcachIndex, int numOfTrying) {
        MemcachedClient memcachedClient;
        int clientSize = memcachedClients.size();
        if ((currentMemcachIndex + numOfTrying) < clientSize) {
            memcachedClient = memcachedClients.get(currentMemcachIndex + numOfTrying);
        } else {
            memcachedClient = memcachedClients.get(currentMemcachIndex + numOfTrying - clientSize);
        }
        return memcachedClient;
    }

    private boolean isMemcachedUseable() {
        if (memcachedClients != null) {
            for (MemcachedClient memcachedClient : memcachedClients) {
                if (memcachedClient.getAvailableServers().size() > 0) {
                    return true;
                }
            }
        } else {
            logger.warn("memcachedClients == " + memcachedClients);
        }
        return false;
    }

    @Override
    public boolean isAvailable() {
        boolean isAvailable = false;
        try {
            this.set(CHECK_AVAILABILITY_KEY, 1000, "OK");
        } catch (Exception e) {
            logger.error("Can't set AVAILABILITY_KEY. ", e);
            return isAvailable;
        }
        Object x = null;
        try {
             x = this.get(CHECK_AVAILABILITY_KEY, 1000);
        } catch (Exception e) {
             logger.error("Can't get AVAILABILITY_KEY. ", e);
             return isAvailable;
        }
        
        if (x!= null && x.equals("OK")) {
            isAvailable = true;
        }
        
        return isAvailable;
    }

    @Override
    public void delete(String key) {
        for (MemcachedClient memcachedClient : memcachedClients) {
            if (memcachedClient.getAvailableServers().size() > 0) {
                try {
                    memcachedClient.delete(key);
                    logger.info("Deleted Key = " + key + " on " + memcachedClient.getAvailableServers());
                } catch (Exception e) {
                    logger.warn("Can't delete (k=" + key + ",v= *** this value has been disabled for a security reason.) to " + memcachedClient.getAvailableServers(), e);
                }
            } else {
                logger.warn("Can't delete (k=" + key + ",v= *** this value has been disabled for a security reason) to " + memcachedClient.getUnavailableServers() + ". Memcached server is not available now.");
            }
        }
    }
}
